package com.demo.ping.component;

import com.demo.ping.orm.entity.ErrorLogs;
import com.demo.ping.orm.enumeration.ResponseEnum;
import com.demo.ping.orm.enumeration.ErrorLevelEnum;
import com.demo.ping.service.ErrorLogsService;
import com.demo.ping.service.KafkaService;
import com.demo.ping.service.PingService;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
import io.micrometer.common.util.StringUtils;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpStatus;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;
import java.util.UUID;

/**
 * @author Jofo
 */
@Slf4j
@Component
public class PingScheduler {

    private final ErrorLogsService errorLogsService;

    private final KafkaService kafkaService;

    private final HazelcastInstance hazelcastInstance;

    private final PingService pingService;

    public PingScheduler(ErrorLogsService errorLogsService, KafkaService kafkaService, @Qualifier("hazelcastInstanceServer") HazelcastInstance hazelcastInstance, PingService pingService) {
        this.errorLogsService = errorLogsService;
        this.kafkaService = kafkaService;
        this.hazelcastInstance = hazelcastInstance;
        this.pingService = pingService;
    }


    private IMap<String, Long> requestTimestamps;

    private static final int MAX_REQUESTS_PER_SECOND = 2;
    private static final String REQUEST_KEY_PREFIX = "requestKey_";

    @PostConstruct
    public void init() {
        requestTimestamps = hazelcastInstance.getMap("requestTimestamps");
    }

    /**
     * Define a timed task method that executes every second
     */
    @Scheduled(fixedRate = 1000)
    public void schedulePing() {
        try {
            // Clean up old requests to maintain the timeliness of the request queue
            cleanUpOldRequests();
            // Obtain the timestamp of the current request to determine whether the maximum number of requests per second is reached
            int currentRequests = requestTimestamps.size();
            // Check whether the current number of requests is less than the maximum number of requests per second
            if (currentRequests < MAX_REQUESTS_PER_SECOND) {
                // If the maximum number of requests is not reached, a ping request is sent
                sendPingRequest();
            } else {
                // If the maximum number of requests is reached, a rate limiting error log is logged
                logRateLimitedError();
            }
        } catch (Exception e) {
            // Catch and log exceptions in the schedulePing method
            log.error("Error in schedulePing: ", e);
        }
    }

    /**
     * Clean up old requests
     * This method is used to clean up requests from a second ago to keep the data structure updated and memory optimized
     * The specific logic is as follows:
     * 1. Gets the timestamp one second before the current time
     * 2. Iterate through the entries with the request timestamp to find all keys with timestamps less than one second ago
     * 3. To prevent the mapping from being modified while traversing, use a separate list to record the keys that need to be removed
     * 4. Finally, iterate through the list to remove these old requests from the original mapping
     */
    private void cleanUpOldRequests() {
        // Gets the timestamp one second before the current time
        long oneSecondAgo = System.currentTimeMillis() - 1000;
        // Initialize a list to store the keys that need to be removed
        List<String> keysToRemove = new ArrayList<>();

        // Iterate through the entries with the request timestamp to find all keys with timestamps less than one second ago
        for (Entry<String, Long> entry : requestTimestamps.entrySet()) {
            if (entry.getValue() < oneSecondAgo) {
                keysToRemove.add(entry.getKey());
            }
        }

        // Iterate through the list of keys that need to be removed to remove these old requests from the original mapping
        for (String key : keysToRemove) {
            requestTimestamps.remove(key);
        }
    }

    /**
     * Sends a ping request to the pingService and logs the request information
     */
    private void sendPingRequest() {
        // Generate a unique request key combining a prefix and the current timestamp to ensure uniqueness
        String requestKey = REQUEST_KEY_PREFIX + System.currentTimeMillis();
        // Generate a unique trace ID for tracking the request
        String traceId = UUID.randomUUID().toString();

        // Send a hello call to the pingService with the trace ID and subscribe to the response
        pingService.sayHello(traceId).subscribe(
                response -> {
                    // record the request timestamp
                    requestTimestamps.put(requestKey, System.currentTimeMillis());
                    // Send a log message to Kafka and log it locally
                    kafkaService.sendMessage("logInfo", ResponseEnum.REQ_SENT_PONG_RESPOND.getReason());
                    log.info(ResponseEnum.REQ_SENT_PONG_RESPOND.getReason());
                },
                error->handlePingError(error, traceId)
        );
    }

    /**
     * Handle errors in pings
     * This method takes different logging policies depending on the type of error
     * In particular, we have handled the case where the response status code is 429 (too many requests).
     */
    private void handlePingError(Throwable error, String traceId) {
        if (error instanceof WebClientResponseException webClientResponseException &&
                webClientResponseException.getStatusCode() == HttpStatus.TOO_MANY_REQUESTS) {
            log.error(ResponseEnum.REQ_SENT_PONG_FAIL.getReason());
            logError(ResponseEnum.REQ_SENT_PONG_FAIL, "schedulePing pingService.handlePingError", error.getMessage(), traceId);
        } else {
            log.error("Connection refused, please check");
            log.error("Connection refused, error: {}", error.getMessage());
        }
    }

    /**
     * this method logs an error message when the maximum number of requests per second is reached
     */
    private void logRateLimitedError() {
        log.error(ResponseEnum.NOT_SENT_BY_RATE_LIMITED.getReason());
        logError(ResponseEnum.NOT_SENT_BY_RATE_LIMITED, "PingScheduler logRateLimitedError", null, null);
    }

    /**
     * Logs an error message to the database and the console
     */
    private void logError(ResponseEnum responseEnum, String stackTrace, String additionalMessage, String traceId) {
        ErrorLogs errorLog = new ErrorLogs();
        errorLog.setErrorLevel(ErrorLevelEnum.MEDIUM.getMessage());
        errorLog.setErrorCode(responseEnum.getCode());
        errorLog.setStackTrace(stackTrace+ (additionalMessage != null ? ": " + additionalMessage : ""));
        errorLog.setErrorMessage(responseEnum.getReason());
        if (StringUtils.isNotBlank(traceId)) {
            errorLog.setTraceId(traceId);
        }
        errorLogsService.saveErrorLog(errorLog);
    }
}
